/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package com.cratos.xxgram.platf.test;

import com.cratos.xxgram.platf.util.Cryptos;
import java.net.CookieHandler;
import java.util.concurrent.*;
import java.net.ProxySelector;
import java.net.URI;
import java.net.URLDecoder;
import java.net.URLEncoder;
import java.net.http.HttpClient;
import java.net.http.WebSocket;
import java.nio.charset.Charset;
import java.security.PublicKey;
import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicInteger;

public class WebSocketDemo {
    public static final String URL= "ws://192.168.0.188:10510/ws/message?bean=%7B'account':'13333333333','password':'96e79218965eb72c92a549dd5a330112','appos':'ios','apptoken':'temptoken'%7D";
//    public static final String URL2 = "ws://192.168.0.188:10510/ws/message?bean=%7B'account':'13044889046','password':'c4ca4238a0b923820dcc509a6f75849b','appos':'ios','apptoken':'temptoken'%7D";
    public static  String message = "{\"message\":{\"module\":\"message\",\"action\":\"onMessage\",\"http_seqno\":\"sasasasa7878\",\"params\":\"bean=%7B%22msgtype%22%3A1001%2C%22targetid%22%3A%22300000004%22%2C%22msgkind%22%3A20%2C%22content%22%3A%22%7Bcontent%3A'xx'%20%7D%22%7D\"}}";
    public static final String xxxmsg = "{\"message\":{\"module\":\"message\",\"action\":\"onMessage\",\"http_seqno\":\"sasasasa7878\",\"params\":\"bean=%s}}";
    public static void main(String[] args) throws InterruptedException {
        String messageData = "{\"msgtype\":1001,\"targetid\":300000004,\"msgkind\":80,\"content\":'%s'}";
        String d = "{content:\\'xx\\' }";
        Cryptos.ECDHCrypto crypto = Cryptos.ECDHCrypto.getInstance();
        String publicKeyStr = "MD4wEAYHKoZIzj0CAQYFK4EEAAkDKgAEkVQBDTsxLnn3PZbZEDtcxCjFFgZ7ezzZCMWDTnHkeukRULrPzu7aoQ==";
        PublicKey readPublicKey = crypto.readPublicKey(publicKeyStr);
        String encryptContent = crypto.encrypt(d, readPublicKey);
        message = String.format(messageData, encryptContent);
        System.out.println("message===" + message);
        HttpClient httpClient = HttpClient.newBuilder()
                .proxy(ProxySelector.getDefault()) //proxy setting
                .connectTimeout(Duration.ofMinutes(1)) // connect timeout setting
                .followRedirects(HttpClient.Redirect.NORMAL) //redirect policy setting 
                .build();
        CompletableFuture<WebSocket> webSocketCompletableFuture = httpClient.newWebSocketBuilder().buildAsync(URI.create(URL),
                new WebSocketClient());
        webSocketCompletableFuture.join();
        batchSend(webSocketCompletableFuture);
        Thread.sleep(10000 * 20);
    }

    public static void batchSend(CompletableFuture<WebSocket> socketCompletableFuture) throws InterruptedException {
        int size = 1;
        BlockingQueue queue = new LinkedBlockingQueue();
        AtomicInteger counter = new AtomicInteger(0);
        CountDownLatch cdl = new CountDownLatch(size);
        ThreadPoolExecutor executor = new ThreadPoolExecutor(3, 6, 1, TimeUnit.MILLISECONDS, queue);
        for (int i = 0; i < size; i++) {
            executor.submit(new Runnable() {
                public void run() {
                    try {
                        socketCompletableFuture.thenCompose(ws -> ws.sendText(message, true)).join();
                        cdl.countDown();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    System.out.println(String.format("thread %d finished counter====>>> %s", this.hashCode(), counter.incrementAndGet()));
                }
            });
        }
        cdl.await();
        System.out.println("执行完毕......" + WebSocketClient.counter.get());
        executor.shutdown();
    }

}
